/// -----------------------------------------------------------------------------------------------------------
/// Module      :  QueueStorageAdapterInboundHandler.cs
/// Description :  This class implements an interface for listening or polling for data.
/// -----------------------------------------------------------------------------------------------------------
/// 

#region Using Directives

using System;
using System.Collections.Generic;
using System.IO;
using System.ServiceModel.Channels;
using System.Xml;

using Microsoft.ServiceModel.Channels.Common;
using Microsoft.WindowsAzure.StorageClient;

using BizTalkFactory.Adapters.AzureStorage.StorageAdapterConstants;

#endregion

namespace BizTalkFactory.Adapters.AzureStorage
{
    internal class AzureStorageAdapterInboundHandler : AzureStorageAdapterHandlerBase, IInboundHandler
    {
        #region Private Members

        private StorageListener listener_ = null;

        private object inboundQueueSynchronizationLock_ = new object();
        private Queue<Message> inboundQueue = new Queue<Message>();

        #endregion

        /// <summary>
        /// Initializes a new instance of the QueueStorageAdapterInboundHandler class
        /// </summary>
        public AzureStorageAdapterInboundHandler(AzureStorageAdapterConnection connection, MetadataLookup metadataLookup)
            : base(connection, metadataLookup)
        {
        }

        #region IInboundHandler Members

        /// <summary>
        /// Start the listener
        /// </summary>
        public void StartListener(string[] actions, TimeSpan timeout)
        {
            // check parameters

            if (actions.Length == 0)
            {
                AzureStorageAdapterConnectionUri connectionUri = Connection.ConnectionFactory.ConnectionUri;
                if (String.IsNullOrEmpty(connectionUri.Path) || (connectionUri.Path == "/"))
                    throw new AdapterException(
                        String.Format(Resources.Err_Invalid_Uri_Path
                        , Connection.ConnectionFactory.ConnectionUri
                        ))
                        ;

                if (Connection.ConnectionFactory.Adapter.PollingStorage == StorageType.Unspecified)
                    throw new AdapterException(
                        String.Format(Resources.Err_Invalid_PollingStorage
                        , Connection.ConnectionFactory.Adapter.PollingStorage
                        ))
                        ;

                switch (Connection.ConnectionFactory.Adapter.PollingStorage)
                {
                    case StorageType.Blob:
                        actions = new string[]{
                         String.Format("{0}{1}/{2}"
                         ,MetadataSearchBrowseNodeIds.BlobNodePrefix
                         ,connectionUri.Path.Substring(1)
                         ,OperationNames.GetBlob)
                        };
                        break;

                    case StorageType.Queue:
                        actions = new string[]{
                            String.Format("{0}{1}/{2}"
                            , MetadataSearchBrowseNodeIds.QueueNodePrefix
                            , connectionUri.Path.Substring(1)
                            , OperationNames.GetMessage)
                        };
                        break;

                    case StorageType.Table:
                        throw new NotImplementedException();
                }
            }

            // initialize listener

            QueueStorageListener queueListener = null;
            BlobStorageListener blobListener = null;

            foreach (string action in actions)
            {
                switch (NodeIdHelper.ParseNodePrefixFromOperationId(action))
                {
                    case MetadataSearchBrowseNodeIds.QueueNodePrefix:
                        {
                            if (queueListener == null)
                            {
                                queueListener = new QueueStorageListener(Connection);
                                queueListener.Received += new StorageListener.StorageEventReceivedEventHandler(queueListener_Received);
                            }

                            string queueName = ParseObjectNameFromOperationId(action);
                            queueListener.QueueNames.Add(queueName);
                        }
                        break;

                    case MetadataSearchBrowseNodeIds.BlobNodePrefix:
                        {
                            if (blobListener == null)
                            {
                                blobListener = new BlobStorageListener(Connection);
                                blobListener.Received += new StorageListener.StorageEventReceivedEventHandler(blobListener_Received);
                            }

                            string containerName = ParseObjectNameFromOperationId(action);
                            blobListener.ContainerName = containerName;
                        }
                        break;
                    default:
                        // TODO: throw
                        break;
                }
            }

            if (queueListener != null)
                listener_ = queueListener;
            if (blobListener != null)
                listener_ = blobListener;

            // enable listeners

            if (listener_ != null)
            {
                listener_.PollingIntervalInSeconds = Connection.ConnectionFactory.Adapter.PollingIntervalInSeconds;
                listener_.EnableRaisingEvents = true;
            }
        }

        void blobListener_Received(object sender, StorageEventArgs e)
        {
            try
            {
                //Create WCF message 
                string action = NodeIdHelper.CreateOperationId(
                    MetadataSearchBrowseNodeIds.BlobNodePrefix
                    , e.StorageName
                    , OperationNames.GetBlob
                    );

                XmlReader reader = XmlReader.Create(
                    SerializeBlobMessage((e as BlobStorageEventArgs).Blob));

                Message message = Message.CreateMessage(
                    MessageVersion.Default
                    , action
                    , reader);

                //store current WCF message into the inbound Queue Message
                lock (inboundQueueSynchronizationLock_)
                    inboundQueue.Enqueue(message);

                // if no errors happened - delete blob from azure storage
                // TODO Add Properties && Metadata to allow user to not delete the blob but update Metadata 
                // to prevent new polling of blob
            }
            catch (Exception /* e */ )
            {
                e.Cancel = true;

                // TODO: handle errors
                // TODO: create message fault
            }
                
            
        }

        private static Stream SerializeBlobMessage(CloudBlob cloudBlob)
        {
            string streamasbase64 = String.Empty;

            using (MemoryStream targetblob = new MemoryStream())
            {
                cloudBlob.DownloadToStream(targetblob);
                targetblob.Seek(0, SeekOrigin.Begin);
                streamasbase64 = Convert.ToBase64String(targetblob.ToArray());
            }


            MemoryStream stream = new MemoryStream();

            XmlWriterSettings settings = new XmlWriterSettings();
            settings.OmitXmlDeclaration = true;
            settings.CloseOutput = false;

            XmlWriter writer = XmlWriter.Create(stream, settings);
            writer.WriteStartElement("GetBlob", AzureStorageAdapter.SERVICENAMESPACE);
            writer.WriteElementString("Content", cloudBlob.DownloadText());
            writer.WriteEndElement();
            writer.Flush();
            writer.Close();

            stream.Seek(0, SeekOrigin.Begin);

            return stream;
        }

        private string ParseObjectNameFromOperationId(string action)
        {
            string objectName = NodeIdHelper.ParseObjectNameFromOperationId(action);
            if (String.IsNullOrEmpty(objectName))
            {
                objectName = Connection.ConnectionFactory.ConnectionUri.Path;
                if (objectName.StartsWith("/")) objectName = objectName.Substring(1);
                if (String.IsNullOrEmpty(objectName))
                    throw new AdapterException(
                        String.Format(Resources.Err_Invalid_Uri_Path
                        , Connection.ConnectionFactory.ConnectionUri
                        ))
                        ;
            }

            return objectName;
        }

        /// <summary>
        /// Stop the listener
        /// </summary>
        public void StopListener(TimeSpan timeout)
        {
            if (listener_ == null)
                return;

            // disable listener

            listener_.EnableRaisingEvents = false;
            listener_ = null;

            // cleanup

            lock (inboundQueueSynchronizationLock_)
                inboundQueue.Clear();
        }

        /// <summary>
        /// Tries to receive a message within a specified interval of time. 
        /// </summary>
        public bool TryReceive(TimeSpan timeout, out System.ServiceModel.Channels.Message message, out IInboundReply reply)
        {
            message = null;
            reply = new QueueStorageAdapterInboundReply();

            TimeoutHelper checkTimeout = new TimeoutHelper(timeout);

            while (true)
            {
                lock (inboundQueueSynchronizationLock_)
                {
                    if (inboundQueue.Count > 0)
                    {
                        message = inboundQueue.Dequeue();
                        if (message != null)
                            return true;
                    }
                }

                if (checkTimeout.IsExpired)
                    return false;

                // yield for next loop

                System.Threading.Thread.Sleep(300);
            }
        }

        /// <summary>
        /// Returns a value that indicates whether a message has arrived within a specified interval of time.
        /// </summary>
        public bool WaitForMessage(TimeSpan timeout)
        {
            TimeoutHelper checkTimeout = new TimeoutHelper(timeout);

            while (true)
            {
                lock (inboundQueueSynchronizationLock_)
                {
                    if ((inboundQueue.Count) > 0 && (inboundQueue.Peek() != null))
                        return true;
                }

                if (checkTimeout.IsExpired)
                    return false;

                // yield for next loop

                System.Threading.Thread.Sleep(300);
            }
        }

        #endregion IInboundHandler Members

        #region Storage Event Handlers Queue Implementation

        void queueListener_Received(object sender, StorageEventArgs e)
        {
            try
            {
                // create WCF message

                string action = NodeIdHelper.CreateOperationId(
                        MetadataSearchBrowseNodeIds.QueueNodePrefix
                        , e.StorageName
                        , OperationNames.GetMessage
                    );

                XmlReader reader = XmlReader.Create(
                    SerializeQueueMessage((e as QueueStorageEventArgs).Message));

                Message message = Message.CreateMessage(
                    MessageVersion.Default
                    , action
                    , reader);

                // store current WCF message into inbound Queue

                lock (inboundQueueSynchronizationLock_)
                    inboundQueue.Enqueue(message);

                // if no errors happened - delete message from Azure Queue

            }
            catch (Exception /* e */)
            {
                e.Cancel = true;

                // TODO: handle errors
                // TODO: create message fault
            }
        }

        private static Stream SerializeQueueMessage(CloudQueueMessage message)
        {
            MemoryStream stream = new MemoryStream();
            
            XmlWriterSettings settings = new XmlWriterSettings();
            settings.OmitXmlDeclaration = true;
            settings.CloseOutput = false;

            XmlWriter writer = XmlWriter.Create(stream, settings);
            writer.WriteStartElement("GetMessage", AzureStorageAdapter.SERVICENAMESPACE);
            writer.WriteStartElement("QueueMessage", AzureStorageAdapter.SERVICENAMESPACE);
            writer.WriteElementString("MessageId", String.Empty, message.Id);
            if (message.InsertionTime.HasValue)
                writer.WriteElementString("InsertionTime", String.Empty, XmlConvert.ToString(message.InsertionTime.Value, XmlDateTimeSerializationMode.Utc));
            if (message.ExpirationTime.HasValue)
                writer.WriteElementString("ExpirationTime", String.Empty, XmlConvert.ToString(message.ExpirationTime.Value, XmlDateTimeSerializationMode.Utc));
            writer.WriteElementString("PopReceipt", String.Empty, message.PopReceipt);
            if (message.NextVisibleTime.HasValue)
                writer.WriteElementString("NextVisibleTime", String.Empty, XmlConvert.ToString(message.NextVisibleTime.Value, XmlDateTimeSerializationMode.Utc));
            writer.WriteElementString("DequeueCount", String.Empty, XmlConvert.ToString(message.DequeueCount));
            writer.WriteElementString("MessageText", String.Empty, message.AsString);
            writer.WriteEndElement();
            writer.WriteEndElement();
            writer.Flush();
            writer.Close();

            stream.Seek(0, SeekOrigin.Begin);

            return stream;
        }

        #endregion
    }

    internal class QueueStorageAdapterInboundReply : InboundReply
    {
        #region InboundReply Members

        /// <summary>
        /// Abort the inbound reply call
        /// </summary>
        public override void Abort()
        {
        }

        /// <summary>
        /// Reply message implemented
        /// </summary>
        public override void Reply(System.ServiceModel.Channels.Message message, TimeSpan timeout)
        {
        }

        #endregion InboundReply Members
    }
}
